package com.ilearn.actor;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

final class Dispatcher {

    private final ActorSystem system;

    private final BlockingQueue<ActorContext> monitorQueue = new LinkedBlockingQueue<ActorContext>();

    private final Thread monitorThread;

    public Dispatcher(final ActorSystem system) {

        this.system = system;
        // 构建一个线程，不断的监听扫描monitorQueue，当ActorSystem不停止，就一直从monitorQueue队列中获取任务执行
        this.monitorThread = new Thread(new Runnable() {

            @Override
            public void run() {
                while (system.status() != ActorSystem.STOP)
                    dispatch();

            }

        }, system.name() + "-monitor");

        this.monitorThread.setDaemon(true);
        this.monitorThread.start();
    }


    // 可以通过ActorSystem的monitor方法给Dispatcher的monitor方法发送Actor
    public void monitor(ActorContext ctx) {
        if (!ctx.isSuspend())
            try {
                monitorQueue.put(ctx);
            } catch (InterruptedException e) {

                e.printStackTrace();
            }
    }

    //
    private void dispatch() {

        try {
            final ActorContext ctx = monitorQueue.poll(120, TimeUnit.SECONDS);

            if (ctx == null) {
                check();
                return;
            }

            if (!ctx.empty() && ctx.lock()) {

                ctx.setDispatchTime(System.currentTimeMillis());

                try {
                    system.executorService().execute(new Runnable() {

                        @Override
                        public void run() {
                            CallBack cb = null;
                            try {
                                Object[] msg = ctx.poll();
                                if (msg != null) {
                                    cb = msg.length == 2 ? (CallBack) msg[1] : null;
                                    ctx.actor().receive(msg[0]);
                                    if (cb != null)
                                        cb.complete(true, null);
                                }

                            } catch (Throwable e) {
                                e.printStackTrace();
                                if (cb != null)
                                    try {
                                        cb.complete(false, e);
                                    } catch (Throwable t) {
                                        t.printStackTrace();
                                    }
                            } finally {
                                ctx.unlock();
                                if (!ctx.empty())
                                    monitor(ctx);
                            }

                        }

                    });
                } catch (Throwable t) {
                    t.printStackTrace();
                    ctx.unlock();
                }

            }


        } catch (InterruptedException e) {

            e.printStackTrace();
        }

    }

    private void check() {

        for (ActorContext ctx : this.system.contexts()) {
            if (ctx.empty() || ctx.isSuspend())
                continue;

            if (ctx.lockTime() > 1000 * 60 * 10) {
                System.err.println(ctx + " lock timeout...");
                //ctx.unlock();
            }

            long lastDispatchTime = ctx.getDispatchTime();
            if (lastDispatchTime == 0 || System.currentTimeMillis() - lastDispatchTime > 1000 * 60) {
                this.monitor(ctx);
            }


        }
    }

    public void shutdown() {
        this.monitorThread.interrupt();
    }

}
